// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: jaewon

#pragma once

#include <memory>
#include <optional>
#include <queue>
#include <string>

#include <boost/asio/ip/tcp.hpp>
#include <boost/beast/core.hpp>
#include <boost/beast/http.hpp>
#include <boost/beast/ssl/ssl_stream.hpp>

#include "coin2/base/time.h"

namespace coin2::base::request {

using tcp = boost::asio::ip::tcp;
namespace ssl = boost::asio::ssl;
namespace beast = boost::beast;
namespace http = boost::beast::http;

class HttpProxyInfo {
 public:
  explicit HttpProxyInfo(const std::string& http_proxy);

  bool UseHttpProxy() const { return use_http_proxy_; }
  bool SslHttpProxy() const { return ssl_http_proxy_; }
  auto HttpProxyAuth() const -> const std::string& { return http_proxy_auth_; }
  auto HttpProxyAddr() const -> const std::string& { return http_proxy_addr_; }
  auto HttpProxyPort() const -> const std::string& { return http_proxy_port_; }

 private:
  void ParseHttpProxy();

 private:
  const std::string http_proxy_;
  bool use_http_proxy_ = false;
  bool ssl_http_proxy_ = false;
  std::string http_proxy_auth_ = "";
  std::string http_proxy_addr_ = "";
  std::string http_proxy_port_ = "0";
};

class HttpsConnection : public std::enable_shared_from_this<HttpsConnection> {
 public:
  class Handler {
   public:
    using RequestType = http::request<http::string_body>;
    using ResponseType = http::response<http::string_body>;

    virtual ~Handler() = default;
    virtual void OnConnected(std::weak_ptr<HttpsConnection> conn) = 0;
    virtual void OnRequestSent(std::weak_ptr<HttpsConnection> conn, const RequestType* req) = 0;
    virtual void OnResponseReceived(
        std::weak_ptr<HttpsConnection> conn,
        const ResponseType* rep) = 0;
    virtual void OnError(std::weak_ptr<HttpsConnection> conn, boost::system::error_code ec) = 0;
  };

  using BufferType = boost::beast::flat_buffer;
  using RequestType = Handler::RequestType;
  using ResponseType = Handler::ResponseType;

  HttpsConnection(
      const std::string& host,
      int port,
      const std::string& bind_ip,
      Handler* handler,
      boost::asio::io_context* io_ctx,
      ssl::context* ssl_ctx,
      const std::string& http_proxy);
  HttpsConnection(
      const std::string& host,
      int port,
      Handler* handler,
      boost::asio::io_context* io_ctx,
      ssl::context* ssl_ctx,
      const std::string& http_proxy);
  virtual ~HttpsConnection();
  // Non-copyable
  HttpsConnection(const HttpsConnection&) = delete;
  HttpsConnection& operator=(const HttpsConnection&) = delete;

  void Start();
  void AsyncSend(const RequestType* req);  // |req| must persist until write.
  void AsyncShutdown();                    // After this is called, no callback will be invoked.
  void AsyncShutdownWithError(boost::system::error_code ec);

  bool is_connected() const { return connected_; }
  bool is_shutdown_requested() const { return shutdown_requested_; }
  size_t num_pending_requests() const { return static_cast<size_t>(in_sending()) + queue_.size(); }
  int64_t age(int64_t curr_ts) const { return curr_ts - created_ts_; }
  int64_t age() const { return age(GetCurrentTimestamp()); }
  bool in_sending() const { return in_flight_request_ != nullptr; }

 private:
  void InitiateConnection();
  void OnResolve(boost::system::error_code ec, tcp::resolver::results_type results);
  void OnConnect(boost::system::error_code ec);
  void OnHandshake(boost::system::error_code ec);
  void OnShutdown(boost::system::error_code ec);

  void AsyncReadNext();
  void AsyncWrite(const RequestType* req);
  void OnRead(boost::system::error_code ec, std::size_t bytes_transferred);
  void OnWrite(boost::system::error_code ec, std::size_t bytes_transferred);
  void OnError(const char* where, boost::system::error_code ec);

  auto& lowest_layer() {
    if (use_ssl_) {
      return beast::get_lowest_layer(*stream_ssl_);
    } else {
      return *stream_nossl_;
    }
  }
  boost::beast::flat_buffer& next_buffer() { return buffer_[(read_count_ + 1) % 2]; }
  ResponseType& next_response() {
    const int idx = (read_count_ + 1) % 2;
    response_[idx] = ResponseType();  // TODO(xguo) better way to clear?
    return response_[idx];
  }
  ResponseType& current_response() { return response_[read_count_ % 2]; }
  void DecorateRequestForHttpProxy(const RequestType* req);

 private:
  boost::asio::io_context* io_ctx_;
  Handler* handler_;
  tcp::resolver resolver_;
  const std::string host_;
  const int port_;
  const std::string bind_ip_;
  bool use_ssl_;
  const bool no_delay_;
  std::unique_ptr<beast::ssl_stream<beast::tcp_stream>> stream_ssl_;
  std::unique_ptr<beast::tcp_stream> stream_nossl_;

  bool connected_ = false;
  bool shutdown_requested_ = false;

  // Double shift buffers
  BufferType buffer_[2];
  ResponseType response_[2];
  int64_t read_count_ = 0;

  // Pending requests queue
  const RequestType* in_flight_request_ = nullptr;
  std::queue<const RequestType*> queue_;

  // Timeout config
  int64_t connect_timeout_ns_ = 5 * 1000000000L;
  // int64_t send_timeout_ns_ = 30 * 1000000000L;
  int64_t shutdown_timeout_ns_ = 5 * 1000000000L;

  // age
  int64_t created_ts_;

  // debug
  int request_sent_num_ = 0;
  int64_t last_request_sent_ts_ = 0;

  // proxy revelant
  std::string http_url_prefix_ = "";
  HttpProxyInfo http_proxy_info_;
};

}  // namespace coin2::base::request
